package offline;

import generator.CounterGenerator;

import java.io.File;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.PriorityQueue;
import java.util.Random;
import java.util.Set;

import tlsb.DBException;
import tlsb.Utils;
import util.Configuration;
import util.IFileWriter;
import util.IPartition;
import util.Pair;
import util.RedisCluster;

/**
 * sourceCluster里面存放的信息： uid的关注和粉丝 新生成的sid包含的用户 uid的分组信息
 * 
 * retCluster包含的信息： uid属于和包含的sid列表
 * 
 * 
 * 对测试时数据的存放有影响的地方： fle, SETOFBELONGEDTL), sid 是以flecid进行partition的 fle ,
 * SETOFOWNDTL, sid是以flecid进行partition的
 * stl的status的存放也是以fstcid存放的，这样可以尽量保证数据就存放在数据本身的节点上。
 * 
 * sid本身的信息还是以sid本身去shard的
 * 
 * 
 * TODO 1. 采用lazy的方式构造curfollowee,而不是一开始就初始化一份。
 * 只有当计算用到某个用户的followwee并且followwee没有计算完才需要写入到redis中。
 * 
 * @author xiafan
 * 
 */
public class CGAXiafan {
	final private String n = "n.txt";
	IFileWriter ifr, ifr0, ifr1, ifr2, ifr3, ifr4, ifr5;
	final public static String SETOFCOMMUNITIES = ":com";
	final private String PREFIXOFSID = "s:";
	final private String SETOFOWNEDTL = ":own";
	final private String SETOFBELONGEDTL = ":blg";
	final private String SETOFCURFOLLOWEES = ":cfle";
	final private String SETOFFOLLOWEES = ":fle";
	final private String ALLUIDS = "uids";
	final private String ALLSIDS = "sids";
	final private String READFREQFIELD = "rf";
	final private String WRITEFREQFIELD = "wf";
	final private String MAXCUTSFIELD = "xcuts";
	final public static String FSTCOMFIELD = "fstc";
	final private String TODOLIST = "todoList";
	final private String PREFIXOFCIDFORSID = "cfs";
	final private int MAX = 999999999;
	final public int T = 10;
	long counter = 0;
	long sidcounter = 0;
	long memcounter = 0;
	long startTime = System.currentTimeMillis();

	// final private String SETOFMEMBERS = ":mem";

	RedisCluster retRedis = null;
	RedisCluster sourceRedis = null;
	IPartition rPart = null;
	IPartition sPart = null;
	Map<Integer, IFileWriter> ifrMap = new HashMap<Integer, IFileWriter>();
	ISetIndex index;
	static Random random = new Random();
	public static final int PORT = 6379;

	String[] hosts = new String[] { "10.11.1.201", "10.11.1.202",
			"10.11.1.203", "10.11.1.204" };

	// String[] hosts = new String[] { "127.0.0.1" };
	// String[] hosts = new String[] { "10.11.1.52", "10.11.1.55", "10.11.1.56"
	// };

	/*
	 * String[] hosts = new String[] { "10.11.1.58", "10.11.1.60", "10.11.1.62",
	 * "10.11.1.63", "10.11.1.65", "10.11.1.66" };
	 */

	CounterGenerator sidsequence;

	public void connect() {

		retRedis = new RedisCluster(hosts, PORT);
		sourceRedis = new RedisCluster(Configuration.instance().getMetaHosts(),
				PORT);
		rPart = retRedis.getPartition();
		sPart = sourceRedis.getPartition();
		index = new InvertedSetIndex(this);
		ifr = new IFileWriter(new File(n), false);
		ifr.open();
		for (int i = 0; i < hosts.length; i++) {
			IFileWriter ifrTemp = new IFileWriter(new File(String.format(
					"n%d.txt", i)), false);
			ifrTemp.open();
			ifrMap.put(i, ifrTemp);
		}
		sidsequence = new CounterGenerator(10000000);
	}

	public void init() throws DBException {
		connect();
		int counter = 0;
		// sidsequence = new CounterGenerator(100000);

		// Set<String> uids = jedis.smembers(ALLUIDS);
		
		// String sid = "";
		// jedis.del(TODOLIST);

		for (String uid : sourceRedis.getJedis(sPart.shard(ALLUIDS)).smembers(
				ALLUIDS)) {
			// 为每个用户b初始化maxcut
			String thisRF = "";
			int maxcuts = MAX;
			// calculate and store the maximum cuts for each user's HTL
			if ((thisRF = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
					READFREQFIELD)) != null) {
				if (thisRF.contains(".")) {
					thisRF = thisRF.substring(0, thisRF.indexOf("."));
				}
				maxcuts = (int) Math.ceil(T
						/ Math.max(1, Integer.parseInt(thisRF)));
			}
			// if one user has no read frequency, make its maxcuts as 10
			// defaultly
			OvlpUnit unit = new OvlpUnit(uid);
			if (maxcuts == MAX)
				maxcuts = 10;
			if (maxcuts > 1) {
				sourceRedis.getPipeline(sPart.shard(uid)).hset(uid,
						MAXCUTSFIELD, Integer.toString(maxcuts));
				sourceRedis.getPipeline(sPart.shard(TODOLIST)).zadd(TODOLIST,
						maxcuts, uid);
				if (counter % 1000 == 0)
					sourceRedis.synAllPipelines();
			} else {
				// 小于1表示timeline不可分割,直接就被丢到shared timeline中去
				unit.finish();
			}
			counter++;
			printcounter(counter, "init");
		}
		sourceRedis.synAllPipelines();
		System.out.println("initialize completed");
	}

	public void computeSTLS() {
		counter = 0;
		while (sourceRedis.getJedis(sPart.shard(TODOLIST)).zcard(TODOLIST) > 0) {
			Set<String> uids = sourceRedis.getJedis(sPart.shard(TODOLIST))
					.zrange(TODOLIST, 0, 0);
			OvlpUnit interUser = null;
			OvlpUnit unit = new OvlpUnit(uids.iterator().next());

			if (unit.getCuts() > 1) {
				intersectWithSTL(unit);
				if (unit.getCuts() > 1) {
					// compute stl in related communities
					// 对于curfolloweelist中还有元素并且curcuts>1的情况，与uid所属的cid中的所有成员做交集
					interUser = intersectWithComUsers(unit);
				}
			}

			if (interUser != null) {
				unit.cleanup();
				interUser.cleanup();
			} else {
				unit.finish();
			}
		}// end 所有用户的关注都已经被切割成STL
	}

	/**
	 * 从已有的STL中选出和uid的followee有交际的那些STL，最后返回uid还可以进行分割的curcut
	 * 
	 * @param uid
	 * @param curFollowees
	 * @return
	 */
	public void intersectWithSTL(OvlpUnit unit) {
		// do if existed stls can be reshared
		long maxOvlp = OVLP_THRESHOLD;

		Map<String, Set<String>> sidToUidMap = index.findSubSetsMap(unit
				.getCurFollowee());
		if (sidToUidMap.size() == 0)
			return;
		StlOvlpCalc calc = new StlOvlpCalc(sidToUidMap, unit);
		calc.init();
		do {
			if (unit.getCuts() <= 1) {
				break;
			} else {
				maxOvlp = OVLP_THRESHOLD;
				Pair<String, Long> reusedSid = calc.nextSid();
				if (reusedSid == null)
					break;
				if (reusedSid.b > maxOvlp) {
					maxOvlp = reusedSid.b;
				} else if (reusedSid.b == 0) {
					break;
				}

				if (maxOvlp > OVLP_THRESHOLD) {
					unit.addToSTL(reusedSid.a, sidToUidMap.get(reusedSid.a));
					calc.delete(reusedSid.a);
					sidToUidMap.remove(reusedSid.a);
				}
			}
		} while (maxOvlp > OVLP_THRESHOLD); // 在subset
											// list中的subset与curfolloweelist没有交集
	}

	/**
	 * 在communities的user中一直进行迭代，知道thisCuts==1。但是目前的实现好像不是这样的，只进行了一次 迭代就结束了 TODO
	 * 性能优化
	 * 
	 * @param uid
	 * @param thisCuts
	 * @param cidList
	 * @return
	 */
	public OvlpUnit intersectWithComUsers(OvlpUnit unit) {
		Iterator<String> fstIter = unit.getProcessedComs().iterator();
		if (!fstIter.hasNext()) {
			return null;
		}

		long maxOvlp = 0;
		OvlpUnit targetU = null;
		Set<String> members = null;
		Map<String, Set<String>> comUsersMap = index.findInterSetMap(unit
				.getCurFollowee());
		for (Entry<String, Set<String>> comUsersEntry : comUsersMap.entrySet()) {
			OvlpUnit newUser = new OvlpUnit(comUsersEntry.getKey());
			if (!belongToComms(newUser, unit.getComs())
					|| comUsersEntry.getKey().equals(unit.getUid()))
				continue;

			long thisOvlp = 0;
			if (comUsersEntry.getValue() != null)
				// 计算交集的写频率
				for (String f : comUsersEntry.getValue()) {
					String thisWF = sourceRedis.getJedis(sPart.shard(f)).hget(
							f, WRITEFREQFIELD);
					if (thisWF != null) {
						if (thisWF.contains(".")) {
							thisWF = thisWF.substring(0, thisWF.indexOf("."));
						}
						thisOvlp += Math.max(0, Integer.parseInt(thisWF));
					}
				}
			if (thisOvlp > maxOvlp) {
				members = comUsersEntry.getValue();
				maxOvlp = thisOvlp;
				targetU = newUser;
			}
		}
		// }
		// do when there is a overlapped user
		if (maxOvlp > 0) {
			// System.err.print("users:"+uid+" "+targetU+" "+members.toString());
			String newsid = targetU.addToSTL(null, members);
			unit.addToSTL(newsid, members);
			return targetU;
		}
		return null;
	}

	private boolean belongToComms(OvlpUnit unit, Set<String> comms) {
		for (String com : unit.getComs()) {
			if (comms.contains(com))
				return true;
		}

		return false;
	}

	public static Set<String> preprocessCommunities(Set<String> cidset) {
		List<String> cidList = new LinkedList<String>();
		int cidcounter = 0;
		for (String cid : cidset) {
			cidcounter++;
			if (cidcounter == 1) {
				cidList.add(cid);
				continue;
			}
			if (cidcounter == 2) {
				if (cid.startsWith(cidList.get(0))) {
					cidList.remove(0);
					cidList.add(cid);
					continue;
				}
				if (!cidList.get(0).startsWith(cid)) {
					cidList.add(cid);
				}
			}
			if (cidcounter == 3) {
				for (String ecid : cidList) {
					if (cid.startsWith(ecid)) {
						cidList.remove(ecid);
					}
				}
				boolean flag = true;
				for (String ecid : cidList) {
					if (ecid.startsWith(cid)) {
						flag = false;
					}
				}
				if (flag) {
					cidList.add(cid);
				}
			}
		}
		return new HashSet<String>(cidList);
	}

	private static final long OVLP_THRESHOLD = 5;

	private class StlOvlpCalc {
		Map<String, Set<String>> sidToUidMap;
		PriorityQueue<Pair<String, Long>> sidQueue;
		OvlpUnit unit;
		Map<String, Set<String>> uidToSidMap = new HashMap<String, Set<String>>();
		Map<String, Pair<String, Long>> sidWfMap = new HashMap<String, Pair<String, Long>>();

		public StlOvlpCalc(Map<String, Set<String>> sidMap_, OvlpUnit unit_) {
			sidToUidMap = sidMap_;
			unit = unit_;
			sidQueue = new PriorityQueue<Pair<String, Long>>(
					sidToUidMap.size() != 0 ? sidToUidMap.size() : 1,
					new Pair.StrLongComparator());
		}

		public void init() {
			long thisOvlp = 0;
			// find out the existing stl that can be reused
			for (Entry<String, Set<String>> entry : sidToUidMap.entrySet()) {
				thisOvlp = calcOvlpWithSTL(unit.getCurFollowee(),
						entry.getValue());
				Pair<String, Long> sidWf = new Pair<String, Long>(
						entry.getKey(), thisOvlp);
				sidQueue.add(sidWf);
				sidWfMap.put(entry.getKey(), sidWf);
				for (String uid : entry.getValue()) {
					if (!uidToSidMap.containsKey(uid)) {
						uidToSidMap.put(uid, new HashSet<String>());
					}
					uidToSidMap.get(uid).add(entry.getKey());
				}

			}
		}

		public Pair<String, Long> nextSid() {
			Pair<String, Long> ret = sidQueue.poll();
			if (ret == null)
				return new Pair<String, Long>("null", 0l);
			return ret;
		}

		public void delete(String sid) {
			Set<String> uids = sidToUidMap.get(sid);
			Set<String> updatedSids = new HashSet<String>();
			for (String u : uids) {
				int wf = getWriteFreq(u);
				for (String updatedSid : uidToSidMap.get(u)) {
					if (!updatedSids.contains(updatedSid)) {
						sidQueue.remove(sidWfMap.get(updatedSid));
						updatedSids.add(updatedSid);
					}
					sidWfMap.get(updatedSid).b = sidWfMap.get(updatedSid).b
							- wf;
				}
			}
			for (String s : updatedSids) {
				sidQueue.add(sidWfMap.get(s));
			}
			sidWfMap.remove(sid);
		}
	}

	private int getWriteFreq(String uid) {
		String thisWF = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
				WRITEFREQFIELD);
		if (thisWF != null) {
			if (thisWF.contains(".")) {
				thisWF = thisWF.substring(0, thisWF.indexOf("."));
			}
			return Math.max(0, Integer.parseInt(thisWF)); // 计算和当前用户交集的写频率
		}
		return 1;
	}

	/**
	 * 计算curFlwee和stl的交集的累加写频率
	 * 
	 * @param curFlwee
	 * @param stl
	 * @return
	 */
	public int calcOvlpWithSTL(Set<String> curFlwee, Set<String> stl) {
		int thisOvlp = 0;
		for (String u : stl) {
			if (curFlwee.contains(u)) {
				String thisWF = sourceRedis.getJedis(sPart.shard(u)).hget(u,
						WRITEFREQFIELD);
				if (thisWF != null) {
					if (thisWF.contains(".")) {
						thisWF = thisWF.substring(0, thisWF.indexOf("."));
					}
					thisOvlp += Math.max(0, Integer.parseInt(thisWF)); // 计算和当前用户交集的写频率
				}
			}
		}
		return thisOvlp;
	}

	/**
	 * members构成一个share timeline，cid是该share timeline对应的cid
	 * 
	 * @param members
	 * @param cid
	 * @return
	 */
	public String addSTL(Set<String> members, String cid) {
		String sid = sourceRedis.getJedis(sPart.shard(members.toString())).get(
				Utils.joinStr("X:", members.toString()));
		if (sid != null) {
			return sid;
		} else {
			// 插入的share timeline不存在时的逻辑
			// sid = Utils.hash(sidsequence.nextInt()) + "";
			sid = sidsequence.nextInt() + "";
			sourceRedis.getJedis(sPart.shard(ALLSIDS)).sadd(ALLSIDS, sid);
			sourceRedis.getJedis(sPart.shard(sid)).set(
					Utils.joinStr(PREFIXOFCIDFORSID, sid), cid); // 存储sid的cid
			ifr.writeln(Utils.joinStr(PREFIXOFCIDFORSID, sid) + "\t" + cid);

			sidcounter++;
			if (sidcounter % 100000 == 0) {
				System.out.println(sidcounter + "sids added");
			}

			// fle的status需要push到sid
			for (String fle : members) {
				memcounter++;
				String flecid = getFirstCommunity(fle);
				// s:001 f1 f2 record uid in an stl
				sourceRedis.getPipeline(sPart.shard((sid))).sadd(
						Utils.joinStr(PREFIXOFSID, sid), fle);
				// 001:blg s1 s2 record stls one user belongs to
				int nid = rPart.shard(flecid);

				retRedis.getPipeline(nid).sadd(
						Utils.joinStr(fle, SETOFBELONGEDTL), sid);
				ifrMap.get(nid).writeln(
						"s\t" + Utils.joinStr(fle, SETOFBELONGEDTL) + "\t"
								+ sid);

			}
			sourceRedis.synAllPipelines();
			retRedis.synAllPipelines();
			String smembers = members.toString();

			// 以sid中的uids为key记录
			sourceRedis.getJedis(sPart.shard(smembers)).set(
					Utils.joinStr("X:", smembers), sid);
			// System.err.println(sid+":"+smembers);
			return sid;
		}
	}

	/**
	 * 把uid的curfollowees作为一个share timeline处理
	 */
	public void moveHTL(OvlpUnit unit) {
		counter++;
		printcounter(counter, "computed users");

		Set<String> members = unit.getCurFollowee();
		if (members.size() > 0) {
			unit.addToSTL(null, members);
		}
	}

	public void cleanup() throws DBException {
		System.out.println("all members in" + sidcounter + " sid:" + counter);
		printSTL();
		ifr.close();
		for (IFileWriter writer : ifrMap.values()) {
			writer.close();
		}
		sourceRedis.close();
		retRedis.close();
	}

	public String getFirstCommunity(String uid) {
		String cid = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
				FSTCOMFIELD);
		while (cid == null) {
			cid = random.nextInt(9999999) + "";
			setFirstCommunity(uid, cid);
			cid = sourceRedis.getJedis(sPart.shard(uid)).hget(uid, FSTCOMFIELD);
			// System.err.println("no fst com");
		}
		return cid;
	}

	public void setFirstCommunity(String uid, String cid) {
		sourceRedis.getJedis(sPart.shard(uid)).hset(uid, FSTCOMFIELD, cid);
		sourceRedis.getJedis(sPart.shard(uid)).sadd(
				Utils.joinStr(uid, SETOFCOMMUNITIES), cid);
		/*
		 * sourceRedis.getPipeline(sPart.shard(cid)).sadd(
		 * Utils.joinStr(PREFIXOFCID, cid), uid);
		 */
		ifr.writeln(FSTCOMFIELD + "\t" + uid + "\t" + cid);
	}

	public RedisCluster getRetRedis() {
		return retRedis;
	}

	public void setRetRedis(RedisCluster retRedis) {
		this.retRedis = retRedis;
	}

	public RedisCluster getSourceRedis() {
		return sourceRedis;
	}

	public void setSourceRedis(RedisCluster sourceRedis) {
		this.sourceRedis = sourceRedis;
	}

	public void printcounter(long counter2, String comment) {
		// counter2++;
		if (counter2 % 10000 == 0) {
			System.out.println(String.format("%f mins elapse;",
					(System.currentTimeMillis() - startTime) / (60000.0)));
			System.out.println(comment + ":" + counter2);
		}

	}

	private static final String COMPUTED_FIELD = "fi";

	private class OvlpUnit {
		final String uid;
		Set<String> curFlwee = null;
		boolean isPersisted = false;
		boolean isComputed = false;
		int cuts = 0;

		Set<String> comms = null;

		public OvlpUnit(String uid) {
			this.uid = uid;

			String computed = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
					COMPUTED_FIELD);
			if (computed != null && computed.equals("1")) {
				isComputed = true;
			} else {
				String mcf = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
						MAXCUTSFIELD);
				if (mcf != null) {
					cuts = Integer.parseInt(mcf);
				}
			}
		}

		public Set<String> getCurFollowee() {
			if (isComputed)
				return new HashSet<String>();

			if (curFlwee == null) {
				curFlwee = sourceRedis.getJedis(sPart.shard(uid)).smembers(
						Utils.joinStr(uid, SETOFCURFOLLOWEES));
			}
			if (curFlwee == null || curFlwee.size() == 0) {
				curFlwee = sourceRedis.getJedis(sPart.shard(uid)).smembers(
						Utils.joinStr(uid, SETOFFOLLOWEES));
			} else {
				isPersisted = true;
			}
			return curFlwee;
		}

		public void persist() {
			if (!isPersisted && curFlwee != null) {
				for (String member : curFlwee) {
					sourceRedis.getPipeline(sPart.shard(uid)).sadd(
							Utils.joinStr(uid, SETOFCURFOLLOWEES), member);
				}
				sourceRedis.synAllPipelines();
				isPersisted = true;
			}
			sourceRedis.getJedis(sPart.shard(uid)).hset(uid, MAXCUTSFIELD,
					Integer.toString(cuts));

			sourceRedis.getJedis(sPart.shard(TODOLIST)).zrem(TODOLIST, uid);
			if (cuts > 1) {
				sourceRedis.getJedis(sPart.shard(TODOLIST)).zadd(TODOLIST,
						cuts, uid);
			}
		}

		public Set<String> getComs() {
			while (comms == null) {
				comms = sourceRedis.getJedis(sPart.shard(uid)).smembers(
						Utils.joinStr(uid, SETOFCOMMUNITIES));
				if (comms == null)
					getFirstCommunity(uid);
			}
			return comms;
		}

		public Set<String> getProcessedComs() {
			return preprocessCommunities(getComs());
		}

		public void cleanup() {
			persist();

			if (cuts <= 1) {
				finish();
			} else if (curFlwee != null && curFlwee.size() == 0) {
				computed();
			}
		}

		public void finish() {
			moveHTL(this);
			computed();
		}

		public void computed() {
			isComputed = true;
			sourceRedis.getJedis(sPart.shard(uid)).hset(uid, COMPUTED_FIELD,
					"1");
			getCurFollowee().clear();
			cuts = 0;

			sourceRedis.getJedis(sPart.shard(uid)).del(
					Utils.joinStr(uid, SETOFCURFOLLOWEES));

			sourceRedis.getJedis(sPart.shard(TODOLIST)).zrem(TODOLIST, uid);
		}

		public void unComputed() {
			isComputed = false;
			sourceRedis.getJedis(sPart.shard(uid)).hset(uid, COMPUTED_FIELD,
					"0");
		}

		public void remElem(Set<String> uids) {
			for (String member : uids) {
				if (isPersisted) {
					sourceRedis.getPipeline(sPart.shard(uid)).srem(
							Utils.joinStr(uid, SETOFCURFOLLOWEES), member);
				}
				sourceRedis.getPipeline(sPart.shard(uid)).srem(
						Utils.joinStr(member, DataLoader.SETOFFOLLOWERS), uid);
			}
			getCurFollowee().removeAll(uids);
			sourceRedis.synAllPipelines();

		}

		public String addToSTL(String sid, Set<String> members) {
			String cid = getFirstCommunity(getUid());
			if (sid == null) {
				sid = addSTL(members, cid);
			}

			// 001:own s1 s2 record stls one user owns
			ifrMap.get(rPart.shard(cid)).writeln(
					"s\t" + Utils.joinStr(getUid(), SETOFOWNEDTL) + "\t" + sid);
			retRedis.getJedis(rPart.shard(cid)).sadd(
					Utils.joinStr(getUid(), SETOFOWNEDTL), sid);
			decreCuts();
			remElem(members);
			return sid;
		}

		public String getUid() {
			return uid;
		}

		public int getCuts() {
			return cuts;
		}

		public void decreCuts() {
			cuts--;
		}
	}

	public void printSTL() {
		Set<String> sidset = sourceRedis.getJedis(sPart.shard(ALLSIDS))
				.smembers(ALLSIDS);

		for (String sid : sidset) {
			System.out.println(sid
					+ " : "
					+ sourceRedis.getJedis(sPart.shard((sid))).smembers(
							Utils.joinStr(PREFIXOFSID, sid)));
		}

		for (String uid : sourceRedis.getJedis(sPart.shard(ALLUIDS)).smembers(
				ALLUIDS)) {
			String cid = sourceRedis.getJedis(sPart.shard(uid)).hget(uid,
					FSTCOMFIELD);
			System.out.println(uid
					+ "~~own:~"
					+ retRedis.getJedis(rPart.shard(cid)).smembers(
							Utils.joinStr(uid, SETOFOWNEDTL)));
			System.out.println(uid
					+ "~~blg:~"
					+ retRedis.getJedis(rPart.shard(cid)).smembers(
							Utils.joinStr(uid, SETOFBELONGEDTL)));

		}
	}

	public static void main(String[] args) throws Exception {

		DataLoader dl = new DataLoader();
		dl.init();
		System.out.println("loading started");
		dl.load();
		System.out.println("loading finished");
		dl.close();

		CGAXiafan cga = new CGAXiafan();

		// cga.connect();

		cga.init();

		cga.computeSTLS();

		cga.cleanup();

	}
}
